/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.common;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.log4j.Logger;

import br.pucrio.inf.lac.konoha.hyuga.events.LocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellContentEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellCoreEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellTransientCheckEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellTransientEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.GridLogicalClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.others.dstream.GridClusterDStream;

/**
 * Class that wraps a set of common Grid routines that are used throughout the EPL rules. For example, it includes 
 * functions, such as {{@link #isNeighbor}, that verify if a cells is neighbor of another one. It also provides a 
 * grid locking and unlocking functionality. This is useful when modifying the cluster list (e.g., on merges).
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class Grid {
    
    /** LOGGER. */
    private static Logger LOGGER = Logger.getLogger(Grid.class.getCanonicalName());
    
    //////////////////////////////////////////////////////////////////////////
    // Locking Functions
    //////////////////////////////////////////////////////////////////////////

    /** Grid Lock */
    private static volatile ReentrantReadWriteLock gridGlobalLock = new ReentrantReadWriteLock(true);
    
    /** Cell's Lock */
    private static volatile Map<Pair<Integer, Integer>, ReentrantLock> gridCellLock = new HashMap<>();
    
    /** 
     * Locks the access to a cell in the grid.
     * 
     * @param x the cell x index
     * @param y the cell y index
     * @return true when it acquire the locks
     */
    public static boolean cellLock(int x, int y) {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " TRYING TO CELL LOCK (" + x + ", " + y + ")");

        Pair<Integer,Integer> cell = Pair.create(x, y);
        
        synchronized (gridCellLock) {
            if (!gridCellLock.containsKey(cell)) {
                gridCellLock.put(cell, new ReentrantLock(true));
            }
        }
        
        gridCellLock.get(cell).lock();
        LOGGER.debug(threadName + " DID CELL LOCK (" + x + ", " + y + ")");
        return true;
    }
    
    /** 
     * Unlocks the access to a cell in the grid.
     * 
     * @param x the cell x index
     * @param y the cell y index
     * @return true when it acquire the locks
     */
    public static boolean cellUnlock(int x, int y) {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " TRYING TO CELL UNLOCK (" + x + ", " + y + ")");
        
        Pair<Integer,Integer> cell = Pair.create(x, y);
        
        synchronized (gridCellLock) {
            gridCellLock.get(cell).unlock();
        }
        
        LOGGER.debug(threadName + " DID CELL UNLOCK (" + x + ", " + y + ")");
        return true;
    }
    
    
    /**
     * Read lock the access to the grid list.
     * 
     * @return true when it acquire the locks
     */
    public static boolean gridReadLock() {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " TRYING TO GRID READ LOCK.");
        
        gridGlobalLock.readLock().lock();

        LOGGER.debug(threadName + " DID GRID READ LOCK.");
        return true;
    }
    
    /**
     * Unlock (read) the access to the grid list.
     * 
     * @param 
     * @return true when it release the locks
     */
    public static boolean gridReadUnlock() {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " TRYING TO GRID READ UNLOCK.");
        
        gridGlobalLock.readLock().unlock();            

        LOGGER.debug(threadName + " DID GRID READ UNLOCK.");
        return true;
    }
    
    /**
     * Write lock the access to the grid list.
     * 
     * @return true when it acquire the locks
     */
    public static boolean gridWriteLock() {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " TRYING TO GRID WRITE LOCK.");
        
        gridGlobalLock.writeLock().lock();

        LOGGER.debug(threadName + " DID GRID WRITE LOCK.");
        return true;
    }
    
    /**
     * Unlock the grid list.
     */
    public static void gridWriteUnlock() {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " TRYING TO GRID WRITE UNLOCK.");

        gridGlobalLock.writeLock().unlock();
        
        LOGGER.debug(threadName + " DID GRID WRITE UNLOCK.");
    }

    /**
     * Check if the grid is write locked.
     * 
     * @return true if the grid is write locked, false otherwise.
     */
    public static boolean isLocked() {
        return gridGlobalLock.isWriteLocked();
    }
    
    /**
     * Reset the grid lock. Useful for debugging and testing.
     */
    public static void gridResetLock() {
        gridCellLock.clear();
        gridGlobalLock = new ReentrantReadWriteLock(true);
    }
    
    //////////////////////////////////////////////////////////////////////////
    // Neighboring Functions
    //////////////////////////////////////////////////////////////////////////
    
    /**
     * Verify if a given cell (x, y) is a neighbor of the cell with (cx, cy) index.
     * 
     * @param x a given cell x index
     * @param y a given cell y index.
     * @param cx the cell x index
     * @param cy the cell y index.
     * @return true if they are neighbor (i.e., they are within 1 index), false otherwise.
     */
    public static boolean isNeighbor(int x, int y, int cx, int cy) {
        int diffx = Math.abs(x - cx);
        int diffy = Math.abs(y - cy);

        if ((diffx <= 1) && (diffy <= 1) && !((diffx == 0) && (diffy == 0))) {
            return true;
        } else {
            return false;
        }
    }
    
    /**
     * Verify if a given cell is an offset neighbor (can be contained) in a giving cluster (set of cells).
     * 
     * @param cell a clustering cell/partition.
     * @param givingCluster a cluster (set of cells/partitions).
     * @return true if they are neighbor (i.e., they are within 1 index), false otherwise.
     */
    public static boolean isNeighborDStream(GridClusterDStream gridCluster, GridClusterDStream otherCluster) {
        boolean neighbor  = false;
        boolean contained = false; // TODO: is contained?

        for (Pair<Integer, Integer> gridIndex : gridCluster.getCoreCells().keySet()) {
            int x = gridIndex.getFirst();
            int y = gridIndex.getSecond();
        
            // Set of cells in givingCluster
            Set<Pair<Integer, Integer>> clusterCells = otherCluster.getCoreCells().keySet();
    
            for (Pair<Integer, Integer> cp : clusterCells) {
                if ((Math.abs(x - cp.getFirst()) <= 1) && (Math.abs(y - cp.getSecond()) <= 1)) {
                    neighbor = true;
                    if ((Math.abs(x - cp.getFirst()) == 0) && (Math.abs(y - cp.getSecond()) == 0)) {
                        contained = true;
                        break;
                    }
                }
            }
        }
        
        if (contained) return false;
        else return neighbor;
    }
    
    /**
     * Inspect if a specific partition (x, y) is contained in a given cluster (set of cells).
     * 
     * @param x x-index of the partition.
     * @param y y-index of the partition.
     * @param cluster a cluster that hold partitions.
     * @return true if the cluster partitions contain the given partition, false otherwise.
     */
    public static boolean containsCell(int x, int y, GridLogicalClusterEvent grid) {
        boolean inside = false;
        
        if (grid.getCoreCells() != null) {
            inside = grid.getCoreCells().contains(new Pair<Integer, Integer>(x, y));
        }

        return inside; 
    }
    
    /**
     * Verify if a given cell is an offset neighbor (can be contained) of a giving cluster (set of cells).
     * 
     * @param cell a CellContentEvent.
     * @param clusterCells a set of pair <x, y> containing the cluster cells (set of partitions).
     * @return true if they are neighbor (i.e., they are within 1 index), false otherwise.
     */
    public static boolean isContainedOrNeighbor(int x, int y, Set<Pair<Integer, Integer>> clusterCells) {
        boolean neighbor = false;

        for (Pair<Integer, Integer> cp : clusterCells) {
            if ((Math.abs(x - cp.getFirst()) <= 1) && (Math.abs(y - cp.getSecond()) <= 1)) {
                neighbor = true;
            }
        }
        return neighbor;
    }
    
    //////////////////////////////////////////////////////////////////////////
    // Grid Functions
    //////////////////////////////////////////////////////////////////////////

    /**
     * Merge a cell cluster event with a set of neighbor clusters.
     * The resulting cluster is formed by the union of the partition cell with all neighbors clusters,
     * since the partition cell will work as a bridge to connect the neighbor clusters.
     *  
     * @param cellCluster the cell cluster to be merged.
     * @param neighborClusters the neighbor clusters.
     * @return a single cluster contained the merged partitions of all clusters and the cell.
     */
    public static Set<Pair<Integer, Integer>> mergeCore(CellClusterEvent cellCluster, GridLogicalClusterEvent[] neighborClusters) {
        String threadName = Thread.currentThread().getName();
        LOGGER.debug(threadName + " MERGE CORE de " + cellCluster);
        Set<Pair<Integer, Integer>> mergedCoreCells = new HashSet<Pair<Integer,Integer>>();
      
        // We always include the cluster partition
        mergedCoreCells.add(new Pair<Integer, Integer>(cellCluster.getCore().getX(), cellCluster.getCore().getY()));
      
        if (neighborClusters != null) {
            // We add the cells of the neighbors cluster
            for (GridLogicalClusterEvent neighbor : neighborClusters) {
                mergedCoreCells.addAll(neighbor.getCoreCells());
            }
        }                
      
        return mergedCoreCells;
    }

    //////////////////////////////////////////////////////////////////////////
    // Casting/Mapping Functions
    //////////////////////////////////////////////////////////////////////////
    
    /**
     * Transform a cell cluster event in a single Grid cluster (that contains this cell).
     * 
     * @param cell the cell clusters
     * @return A grid cluster containg the cell 
     */
    public static GridLogicalClusterEvent toGridCluster(CellClusterEvent cell) {
        return new GridLogicalClusterEvent(new CellClusterEvent[]{cell});
    }
    
    /**
     * A map function. This function is called if the sum of the transient and its neighboring density, considering the
     * decay factor, surpass minPts. It transforms a transient, but dense, cell event in a {@link CellClusterEvent} that
     * contains the transient cell and its neighboring cells. Thus, the transient cell is reintroduced equally to a
     * dense cell event in the EPN flow.
     * 
     * @param transientCell the combination of the transient cell and its neighboring cells.
     * @returns A dense {@link CellClusterEvent}.

     */
    public static CellClusterEvent toCellCluster(CellTransientCheckEvent transientCell) {
        CellTransientEvent cell = transientCell.getCore();
        CellCoreEvent cellCore = new CellCoreEvent(cell.getX(), cell.getY(), cell.getNodes());
        
        return new CellClusterEvent(cellCore, transientCell.getNeighbors());
    }

    //////////////////////////////////////////////////////////////////////////
    // Decay Function
    //////////////////////////////////////////////////////////////////////////

    /** Decay function to use (true = linear, false = exponential). */
    private static boolean linearDecay = true;
    
    /** Decay factor when using exponential decay. */
    private static double expdecayfactor = Integer.MIN_VALUE;
    
    public static void setDecayFunction(String functionToUse) {
        if (functionToUse.equalsIgnoreCase("exponential") || functionToUse.equalsIgnoreCase("exp")) {
            linearDecay = false;
        } else {
            linearDecay = true;
        }
    }
    
    /**
     * This function sums the number of mobile nodes in the neighbors cell subdivision, considering discrete intervals,
     * to verify if a transient cell is dense. Specifically, it applies a weight based on the number of cell divisions
     * and sum these values (of the neighboring cells) with the transient cell density.
     *   
     * @param eps DBSCAN ε distance.
     * @param minPts DBSCAN minPts minimum number of point. 
     * @param cellSubDiv the number of cell subdivisions.
     * @param core the core cell to check
     * @param neighbors the core cell neighborhood
     * @return true if the density is above minPts, false otherwise.
     */
    public static boolean isDense(double eps, int minPts, int cellSubDiv, CellContentEvent core, CellContentEvent[] neighbors) {
        double currentDensity = core.size();
        if (currentDensity >= minPts) return true;
        if (cellSubDiv <= 0)          return currentDensity >= minPts;
        
        int x = core.getX();
        int y = core.getY();
        
        // Check Neighboring Cells (first Von Neumann neighborhood, then Moore one)
        if (neighbors != null) {
            int nx = 0;
            int ny = 0;
            for (CellContentEvent neighborCell : neighbors) {
                nx = neighborCell.getX();
                ny = neighborCell.getY();
                
                // Von Neumann neighborhood
                if  ((nx == x - 1) && (ny == y)) {       // Left
                    currentDensity+= getDensity(cellSubDiv, reverse(getCellSubdivision(cellSubDiv, true, neighborCell)));
                } else if ((nx == x + 1) && (ny == y)) { // Right
                    currentDensity+= getDensity(cellSubDiv, getCellSubdivision(cellSubDiv, true, neighborCell));
                } else if ((nx == x) && (ny == y + 1)) { // Top
                    currentDensity+= getDensity(cellSubDiv, getCellSubdivision(cellSubDiv, false, neighborCell));
                } else if ((nx == x) && (ny == y - 1)) { // Bottom
                    currentDensity+= getDensity(cellSubDiv, reverse(getCellSubdivision(cellSubDiv, false, neighborCell)));
                // Moore neighborhood (diagonals)
                } else if ((nx == x - 1) && (ny == y + 1)) { // Top Left
                    currentDensity+= getDensity(cellSubDiv, getDiagonalCellSubdivision(cellSubDiv, true, false,  neighborCell));
                } else if ((nx == x - 1) && (ny == y - 1)) { // Bottom Left
                    currentDensity+= getDensity(cellSubDiv, getDiagonalCellSubdivision(cellSubDiv, true, true,   neighborCell));
                } else if ((nx == x + 1) && (ny == y + 1)) { // Top Right
                    currentDensity+= getDensity(cellSubDiv, getDiagonalCellSubdivision(cellSubDiv, false, false, neighborCell));
                } else if ((nx == x + 1) && (ny == y - 1)) { // Bottom Right
                    currentDensity+= getDensity(cellSubDiv, getDiagonalCellSubdivision(cellSubDiv, false, true,  neighborCell));
                }  
                
                if (currentDensity > minPts) break;
            }
        }
        
        return currentDensity > minPts;
    }
    
    /** 
     * A basic decay (linear) function that associate a cell subdivision distribution with a density value.
     * 
     * First, the function calculate the weight of a single slot (subdivision = 1 / numberOfCellSubDivisions).
     * Then, it assigns a decreasing weight for each subdivision as following. 
     * Suppose the neighboring cell have four subdivisions:
     * <pre> 
     *  [ TRANSIENT CELL ] [ 1 | 2 | 3 | 4 ]
     * </pre>
     * The weight of the mobile nodes in the first subdivision (closer to the transient cell) is: 1 - weight.
     * In the second subdivision the weight is 1 - 2 × weight, and so on. Thus, the farther the subdivision slot is from
     * the transient cell the weaker its weight is.
     * Finally, the density is calculated as the number of mobile nodes in a subdivision times this weight. 
     * Thus, the final weight is:
     * <pre>
     *   n
     *   ∑ (1 - i × weight) × subdivision[i]
     *  i=1
     * </pre>
     * 
     * @param numberCellSubDiv the number of subdivisions (n) in a cell
     * @param subDivArray an array that contain the number of mobile nodes in each subdivision
     * @return double the density value of this cell.
     */
    private static double getLinearDensity(int numberCellSubDiv, int[] subDivArray) {
        double density = 0;
        double factor  = (double) 1 / (double) numberCellSubDiv;
        
        for (int i = 0; i < subDivArray.length; i++) {
            density += (1 - ((i+1) * factor)) * subDivArray[i]; 
        }
        
        return density;
    }
    
    private static double getDecayFactor(int numberCellSubDiv) {
        if (expdecayfactor == Integer.MIN_VALUE) {
            // x^n = 0.01
            // ln x^n = ln 0.01
            // n * lnx = ln 0.01
            // ln x = ln 0.01 / n
            double zerolog  = Math.log(0.01);
            double decaylog = zerolog / numberCellSubDiv;
            expdecayfactor = Math.pow(Math.E, decaylog);
        }
        
        return expdecayfactor;
    }
    
    private static double getExponenetialDensity(int numberCellSubDiv, int[] subDivArray) {
        double density = 0;
        double factor = getDecayFactor(numberCellSubDiv);
        
        for (int i = 0; i < subDivArray.length; i++) {
            density += Math.pow(factor, i) * subDivArray[i]; 
        }
        
        return density;
    }
    
    private static double getDensity(int numberCellSubDiv, int[] subDivArray) {
        if (linearDecay) {
            return getLinearDensity(numberCellSubDiv, subDivArray);
        } else {
            return getExponenetialDensity(numberCellSubDiv, subDivArray);
        }
    }
    
    /**
     * This function returns an array that contains the distribution of mobile nodes inside a neighboring cell for each
     * cell subdivision, that is, the first value of the resulting array contains the sum of mobile nodes inside the 
     * first slot, while the second value contains the sum of the second slot, and so on.
     * 
     * <pre>
     *                   ----------------- 
     *                   |       4       |
     *                   ----------------- 
     *                   |       3       |
     *                   -----------------
     *                   |       2       |
     *                   -----------------
     *                   |       1       |
     *                   -----------------   
     * ----------------- ----------------- ----------------- 
     * | 4 | 3 | 2 | 1 | | TRANSIENTCELL | | 1 | 2 | 3 | 4 |
     * ----------------- ----------------- ----------------- 
     *                   ----------------- 
     *                   |       1       |
     *                   ----------------- 
     *                   |       2       |
     *                   -----------------
     *                   |       3       |
     *                   -----------------
     *                   |       4       |
     *                   ----------------- 
     * </pre>
     * @param numberCellSubDiv The number of subdivision a cell contain.
     * @param horizontalsweep A parameter that indicate the sweep technique to use (true for horizontal, false for vertical)
     * @param cell A cell and its location updates
     * @return an int array that contains the sum of the number of mobile nodes in each of the cell subdivison slot.
     */
    private static int[] getCellSubdivision(int numberCellSubDiv, boolean horizontalsweep, CellContentEvent cell) {
        int[] subDivArray = new int[numberCellSubDiv];
        
        LocationUpdateEvent[] nodes = cell.getNodes();
        for (LocationUpdateEvent lu : nodes) {
            if (horizontalsweep)
                subDivArray[lu.getHsweep()]++;
            else
                subDivArray[lu.getVsweep()]++;
        }
        
        return subDivArray;
    }
    
    /**
     * This function returns an array that contains the distribution of mobile nodes inside a diagonal neighboring cell
     * for each cell subdivision.
     * 
     * To do this, we join the horizontal and vertical subdivision to discover the intersection cells, as shown in
     * the following configuration:
     * <pre> 
     * -----------------                   -----------------
     * | 4 | 4 | 4 | 4 |                   | 4 | 4 | 4 | 4 |
     * -----------------                   -----------------
     * | 4 | 3 | 3 | 3 |                   | 3 | 3 | 3 | 4 |
     * -----------------                   -----------------
     * | 4 | 3 | 2 | 2 |                   | 2 | 2 | 3 | 4 |
     * -----------------                   -----------------
     * | 4 | 3 | 2 | 1 |                   | 1 | 2 | 3 | 4 |
     * -----------------                   -----------------
     *                  ------------------
     *                  | TRANSIENT CELL |
     *                  ------------------
     * -----------------                   -----------------
     * | 4 | 3 | 2 | 1 |                   | 1 | 2 | 3 | 4 |
     * -----------------                   -----------------
     * | 4 | 3 | 2 | 2 |                   | 2 | 2 | 3 | 4 |
     * -----------------                   -----------------
     * | 4 | 3 | 3 | 3 |                   | 3 | 3 | 3 | 4 |
     * -----------------                   -----------------
     * | 4 | 4 | 4 | 4 |                   | 4 | 4 | 4 | 4 |
     * -----------------                   -----------------                  
     *  </pre>
     *  
     *  Mobile nodes inside the "1" subdivision are 1 distance away from the transient cell, while those inside the "2"
     *  are 2 distance away and so on. The returning array contain the sum of mobile nodes inside each cell subdivision.
     * 
     * @param numberCellSubDiv The number of subdivision a cell contain.
     * @param reverseHorizontal Do we need to revert the horizontal index? (Useful for diagonal cells)
     * @param reverseVertical Do we need to revert the vertical index? (Useful for diagonal cells)
     * @param horizontalSubdivision The horizontal subdivision of the mobile nodes inside the cell
     * @param verticalSubdivision The vertical subdivision of the mobile nodes inside the cell
     * @return an int array that contains the sum of the number of mobile nodes in each of the cell subdivision slot.
     */
    private static int[] getDiagonalCellSubdivision(int numberCellSubDiv, boolean reverseHorizontal, 
                                                    boolean reverseVertical, CellContentEvent cell) {
        int[] subDivArray = new int[numberCellSubDiv];

        LocationUpdateEvent[] nodes = cell.getNodes();
        for (LocationUpdateEvent lu : nodes) {
            int hindex = 0;
            int vindex = 0;
            
            if (!reverseHorizontal && !reverseVertical) {
                hindex = lu.getHsweep();
                vindex = lu.getVsweep();
            } else if (reverseHorizontal && !reverseVertical) {
                hindex = numberCellSubDiv - lu.getHsweep() - 1;
                vindex = lu.getVsweep();
            } else if (!reverseHorizontal && reverseVertical) {
                hindex = lu.getHsweep();
                vindex = numberCellSubDiv - lu.getVsweep() - 1;
            } else if (reverseHorizontal && reverseVertical) {
                hindex = numberCellSubDiv - lu.getHsweep() - 1;
                vindex = numberCellSubDiv - lu.getVsweep() - 1;
            }

            subDivArray[Math.max(hindex, vindex)]++;
        }
        
        return subDivArray;
    }
    
    /**
     * Reverse an integer array.
     * 
     * @param inArray the input array to reverse.
     * @return the reversed array.
     */
    private static int[] reverse(int[] inArray) {
        int length = inArray.length;
        int[] rArray = new int[length];
        
        for (int i = 0; i < length; i++) {
            rArray[i] = inArray[length - 1 - i];
        }
        
        return rArray;
    }

    //////////////////////////////////////////////////////////////////////////
    // Util Functions
    //////////////////////////////////////////////////////////////////////////

    public static void reset() {
        gridCellLock.clear();
        gridGlobalLock = new ReentrantReadWriteLock(true);
    }
    
    public static boolean debug() {
        return true;
    }    

	
}
